package com.shujia.mr.mapJoin;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

/**
 * 需求：
 * 将学生的各科成绩和学科信息进行关联
 * <p>
 * 分析：
 * 1.读取大表中的数据
 * 2.读取Object.txt中的数据
 *
 * 注意：
 *      使用MapJoin前提为小表和大表关联，一般来说一个表中的数据量在25M以下认为是小表
 *          如果小表过大，那么在MapTask中容易造成内存溢出错误
 *
 */
public class MapJoinMapper extends Mapper<LongWritable, Text, NullWritable, Text> {

    HashMap<String, String> objectIDAndName = new HashMap<>();

    /**
     * 在setup中读取Object.txt中的数据并加载至内存当中
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Mapper<LongWritable, Text, NullWritable, Text>.Context context) throws IOException, InterruptedException {
        // 该函数在任务提交时，被执行一次
//        System.out.println("MapJoinMapper类中的setUp函数被执行...");
        URI[] cacheFiles = context.getCacheFiles();
        URI cacheFile = cacheFiles[0];
        Configuration conf = context.getConfiguration();
        FileSystem fileSystem = FileSystem.get(conf);

        // 通过fileSystem创建InputStream 流对象，读取文件时 为了方便选择使用bufferedReader
        //  创建bufferedReader时需要给定Reader的子类对象，其中有InputStreamReader该类可以将InputStream和Reader连接
        FSDataInputStream fsDataInputStream = fileSystem.open(new Path(cacheFile));
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(fsDataInputStream));

        String line;
        while ((line = bufferedReader.readLine()) != null) {
            // 1000001,数学
            String[] split = line.split(",");
            objectIDAndName.put(split[0],split[1]);
        }

        // 关闭流
        bufferedReader.close();
        fsDataInputStream.close();
        fileSystem.close();

    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, NullWritable, Text>.Context context) throws IOException, InterruptedException {
        // 1500100001,1000002,5
        String[] columns = value.toString().split(",");
        // 通过科目ID 获取科目的名称
        String name = objectIDAndName.get(columns[1]);

        context.write(NullWritable.get(),new Text(columns[0]+","+columns[1]+","+columns[2]+","+name));


//        context.write(NullWritable.get(), value);
    }
}
